package com.db.priceApp.network;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Logger;

import com.db.priceApp.domain.clientDataBasa.ClientsBasa;
import com.db.priceApp.network.protocol.Message;
import com.db.priceApp.network.protocol.MessageProcessor;
import com.db.priceApp.network.protocol.MessageTranciever;

public class Server implements Runnable {
	private volatile ServerSocketChannel ssc;
	private volatile ServerSocket ss;
	private volatile Selector selector;
	private AtomicBoolean serverIsStoped;
	private final BlockingQueue<Message> clientMarketDataRequests;
	private final ClientsBasa clientBasa;
	private Logger logger = Logger.getLogger(Server.class.getName());

	public Server(BlockingQueue<Message> clientMessages, ClientsBasa basa) {
		clientMarketDataRequests = clientMessages;
		clientBasa = basa;
		serverIsStoped = new AtomicBoolean(false);
		connectionInitialization();
	}

	private void connectionInitialization() {
		try {
			selector = Selector.open();
			ssc = ServerSocketChannel.open();
			ss = ssc.socket();
			ss.bind(new InetSocketAddress(1027));

			ssc.configureBlocking(false);
			ssc.register(selector, SelectionKey.OP_ACCEPT);
		} catch (IOException e) {
			logger.severe("fail to open a connection");
		}
	}

	public void run() {
		try {
			while (!serverIsStoped.get()) {
				// wait for incoming events (new connections or orders)
				selector.select();
				Set<SelectionKey> selectedKeys = selector.selectedKeys();
				for (SelectionKey skey : selectedKeys) {

					if (skey.isAcceptable()) { // get a new connection

						setNewConnection(skey);

					} else if (skey.isReadable()) { // get new message

						try {
							readMessageFromChannel(skey);
						} catch (IOException e) { // unset connection
							clientBasa.removeClient(skey);							
							skey.cancel();
						}
					}
				}
				selectedKeys.clear();
			}
		} catch (IOException e) {
			logger.severe("server failure: " + e.getMessage());
			throw new RuntimeException("Server error", e);
		} finally {
			
			try {
				ss.close();
				ssc.close();
				selector.close();
			} catch (IOException e) {			
				logger.severe("failed to shut down properly");
			}
			serverIsStoped.set(true);
		}
	}

	private void setNewConnection(SelectionKey skey) throws IOException,
			ClosedChannelException {
		ssc = (ServerSocketChannel) skey.channel();
		SocketChannel sc = ssc.accept();
		// register a new connection with selector
		sc.configureBlocking(false);
		sc.register(selector, SelectionKey.OP_READ);
	}

	private void readMessageFromChannel(SelectionKey skey) throws IOException {

		byte[] clientOrder;
		Message clientMessage = new Message();
		MessageProcessor mp = new MessageProcessor();
		MessageTranciever mt = new MessageTranciever();

		mt.receive((SocketChannel) skey.channel());

		while (mt.hasMessage()) {
			try {
				clientOrder = mt.nextMessage();
				clientMessage = mp.processMessage(clientOrder,
						clientBasa.getLogin(skey), skey);
				logger.info(clientMessage.toString());
				// add client's message in queue to process
				clientMarketDataRequests.put(clientMessage);
			} catch (IllegalArgumentException e) {
				logger.warning("Illegal message format: " + e.getMessage());
				continue;
			} catch (InterruptedException e) {
				logger.warning("fail to process a client request: " + clientMessage.toString());
				continue;
			}
		}
	}

}
